/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.store.ha;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.common.ServiceThread;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.remoting.common.RemotingUtil;
import com.alibaba.rocketmq.store.CommitLog.GroupCommitRequest;
import com.alibaba.rocketmq.store.DefaultMessageStore;

/**
 * HA服务，负责同步双写，异步复制功能
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-21
 */
public class HAService {
	private static final Logger log = LoggerFactory.getLogger(LoggerName.StoreLoggerName);
	// 客户端连接计数
	private final AtomicInteger connectionCount = new AtomicInteger(0);
	// 存储客户端连接
	private final List<HAConnection> connectionList = new LinkedList<HAConnection>();
	// 接收新的Socket连接
	private final AcceptSocketService acceptSocketService;
	// 顶层存储对象
	private final DefaultMessageStore defaultMessageStore;
	// 异步通知
	private final WaitNotifyObject waitNotifyObject = new WaitNotifyObject();
	// 写入到Slave的最大Offset
	private final AtomicLong push2SlaveMaxOffset = new AtomicLong(0);
	// 主从复制通知服务
	private final GroupTransferService groupTransferService;
	// Slave订阅对象
	private final HAClient haClient;

	public HAService(final DefaultMessageStore defaultMessageStore) throws IOException {
		this.defaultMessageStore = defaultMessageStore;
		// 该服务是主用Broker使用，该服务主要监听新的Socket连接，若有新的连接到来，则创建HAConnection对象，
		// 在该对象中创建了HAConnection.WriteSocketService和HAConnection.ReadSocketService线程服务，对新来的socket连接分别进行读和写的监听
		this.acceptSocketService = new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
		// 该服务是对同步进度进行监听，若达到应用层的写入偏移量，则通知应用层该同步已经完成。
		// 在调用CommitLog.putMessage方法写入消息内容时，根据主用broker的配置来决定是否利用该服务进行同步等待数据同步的结果。
		this.groupTransferService = new GroupTransferService();
		// 该服务是备用Broker使用，在备用Broker启动之后与主用Broker建立socket连接，
		// 然后将备用Broker的commitlog文件的最大数据位置每隔5秒给主用Broker发送一次；并监听主用Broker的返回消息
		this.haClient = new HAClient();
	}

	public void updateMasterAddress(final String newAddr) {
		if (this.haClient != null) {
			this.haClient.updateMasterAddress(newAddr);
		}
	}

	public void putRequest(final GroupCommitRequest request) {
		this.groupTransferService.putRequest(request);
	}

	/**
	 * 判断主从之间数据传输是否正常
	 * 
	 * @return
	 */
	// 主用put的位置masterPutwhere等于wroteOffset（写入的开始物理位置）加上wroteBytes（写入的大小）
	public boolean isSlaveOK(final long masterPutWhere) {
		boolean result = this.connectionCount.get() > 0;
		result = result && ((masterPutWhere - this.push2SlaveMaxOffset.get()/* 写入到Slave的最大Offset */) < this.defaultMessageStore.getMessageStoreConfig().getHaSlaveFallbehindMax());
		return result;
	}

	/**
	 * 通知复制了部分数据
	 * 若slaveAckOffset大于HAService.push2SlaveMaxOffset的值则更新push2SlaveMaxOffset的值，
	 * 并通知调用GroupTransferService.notifyTransferSome方法唤醒GroupTransferService服务线程。
	 * 在同步双写模式下面，前端调用者会通过此线程服务来监听同步进度情况。
	 */
	public void notifyTransferSome(final long offset) {
		for (long value = this.push2SlaveMaxOffset.get(); offset > value;) {
			boolean ok = this.push2SlaveMaxOffset.compareAndSet(value, offset);
			if (ok) {
				this.groupTransferService.notifyTransferSome();
				break;
			} else {
				value = this.push2SlaveMaxOffset.get();
			}
		}
	}

	public AtomicInteger getConnectionCount() {
		return connectionCount;
	}

	// public void notifyTransferSome() {
	// this.groupTransferService.notifyTransferSome();
	// }

	public void start() {
		this.acceptSocketService.beginAccept();
		this.acceptSocketService.start();
		this.groupTransferService.start();
		this.haClient.start();
	}

	public void addConnection(final HAConnection conn) {
		synchronized (this.connectionList) {
			this.connectionList.add(conn);
		}
	}

	public void removeConnection(final HAConnection conn) {
		synchronized (this.connectionList) {
			this.connectionList.remove(conn);
		}
	}

	public void shutdown() {
		this.haClient.shutdown();
		this.acceptSocketService.shutdown(true);
		this.destroyConnections();
		this.groupTransferService.shutdown();
	}

	public void destroyConnections() {
		synchronized (this.connectionList) {
			for (HAConnection c : this.connectionList) {
				c.shutdown();
			}

			this.connectionList.clear();
		}
	}

	public DefaultMessageStore getDefaultMessageStore() {
		return defaultMessageStore;
	}

	public WaitNotifyObject getWaitNotifyObject() {
		return waitNotifyObject;
	}

	/**
	 * 该服务是主用Broker使用，该服务主要监听新的Socket连接，若有新的连接到来，则创建HAConnection对象，
	 * 在该对象中创建了HAConnection.WriteSocketService和HAConnection.ReadSocketService线程服务，对新来的socket连接分别进行读和写的监听
	 * @author Administrator
	 *
	 */
	class AcceptSocketService extends ServiceThread {
		private ServerSocketChannel serverSocketChannel;
		private Selector selector;
		private SocketAddress socketAddressListen;

		public AcceptSocketService(final int port) {
			this.socketAddressListen = new InetSocketAddress(port);
		}

		public void beginAccept() {
			try {
				this.serverSocketChannel = ServerSocketChannel.open();
				this.selector = RemotingUtil.openSelector();
				this.serverSocketChannel.socket().setReuseAddress(true);
				this.serverSocketChannel.socket().bind(this.socketAddressListen);
				this.serverSocketChannel.configureBlocking(false);
				this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
			} catch (Exception e) {
				log.error("beginAccept exception", e);
			}
		}

		@Override
		public void run() {
			log.info(this.getServiceName() + " service started");

			while (!this.isStoped()) {
				try {
					this.selector.select(1000);
					Set<SelectionKey> selected = this.selector.selectedKeys();
					if (selected != null) {
						for (SelectionKey k : selected) {
							if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
								// 如果有Accept事件，就创建一个SocketChannel，并注册SelectionKey_OP_READ
								// 若有该操作达到，即有备用Broker请求连接到该主用Broker上时
								SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();
								if (sc != null) {
									HAService.log.info("HAService receive new connection, " + sc.socket().getRemoteSocketAddress());

									try {
										HAConnection conn = new HAConnection(HAService.this, sc);
										// 该方法中启动该对象的ReadSocketService和WriteSocketService服务线程
										// ReadSocketService服务线程用于读取备用Broker的请求，WriteSocketService服务线程用于向备用Broker写入数据
										conn.start();
										// 将该HAConnection对象存入HAService.connectionList:List<HAConnection>变量中，
										// 该变量是存储客户端连接，用于管理连接的删除和销毁
										HAService.this.addConnection(conn);
									} catch (Exception e) {
										log.error("new HAConnection exception", e);
										sc.close();
									}
								}
							} else {
								log.warn("Unexpected ops in select " + k.readyOps());
							}
						}

						selected.clear();
					}

				} catch (Exception e) {
					log.error(this.getServiceName() + " service has exception.", e);
				}
			}

			log.error(this.getServiceName() + " service end");
		}

		@Override
		public String getServiceName() {
			return AcceptSocketService.class.getSimpleName();
		}
	}

	/**
	 * 该服务是对同步进度进行监听，若达到应用层的写入偏移量，则通知应用层该同步已经完成。
	 * 在调用CommitLog.putMessage方法写入消息内容时，根据主用broker的配置来决定是否利用该服务进行同步等待数据同步的结果。
	 * GroupTransferService Service
	 */
	class GroupTransferService extends ServiceThread {
		// 异步通知
		private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject();
		private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>();
		private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>();

		public void putRequest(final GroupCommitRequest request) {
			synchronized (this) {
				this.requestsWrite.add(request);
				if (!this.hasNotified) {
					this.hasNotified = true;
					this.notify();

					// TODO 这里要Notify两个线程 1、GroupTransferService
					// 2、WriteSocketService
					// 在调用putRequest后，已经Notify了WriteSocketService
				}
			}
		}

		public void notifyTransferSome() {
			this.notifyTransferObject.wakeup();
		}

		private void swapRequests() {
			List<GroupCommitRequest> tmp = this.requestsWrite;
			this.requestsWrite = this.requestsRead;
			this.requestsRead = tmp;
		}

		private void doWaitTransfer() {
			if (!this.requestsRead.isEmpty()) {
				for (GroupCommitRequest req : this.requestsRead) {
					boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
					for (int i = 0; !transferOK && i < 5;) {
						this.notifyTransferObject.waitForRunning(1000);
						transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
					}

					if (!transferOK) {
						log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
					}
					// 若push2SlaveMaxOffset大于了该对象的NextOffset值则置GroupCommitRequest请求对象的flushOK变量为true，在调用处对该变量有监听
					req.wakeupCustomer(transferOK);
				}

				this.requestsRead.clear();
			}
		}

		public void run() {
			log.info(this.getServiceName() + " service started");

			while (!this.isStoped()) {
				try {
					// 将requestsWrite队列的请求与requestsRead队列的请求交换，而requestsRead队列一般都是空的，也就是将requestsWrite队列的内容赋值到requestsRead队列之后再清空
					this.waitForRunning(0);
					this.doWaitTransfer();
				} catch (Exception e) {
					log.warn(this.getServiceName() + " service has exception. ", e);
				}
			}

			log.info(this.getServiceName() + " service end");
		}

		@Override
		protected void onWaitEnd() {
			this.swapRequests();
		}

		@Override
		public String getServiceName() {
			return GroupTransferService.class.getSimpleName();
		}
	}

	/**
	 * 该服务是备用Broker使用，
	 *1、 在备用Broker启动之后与主用Broker建立socket连接，然后将备用Broker的commitlog文件的最大数据位置每隔5秒给主用Broker发送一次；
	 *2、 监听主用Broker的返回消息，然后进行后续处理
	 * @author Administrator
	 *
	 */
	class HAClient extends ServiceThread {
		private static final int ReadMaxBufferSize = 1024 * 1024 * 4; // 4G
		// 主节点IP:PORT
		private final AtomicReference<String> masterAddress = new AtomicReference<String>();
		// 向Master汇报Slave最大Offset
		private final ByteBuffer reportOffset = ByteBuffer.allocate(8);
		private SocketChannel socketChannel;
		private Selector selector;
		private long lastWriteTimestamp = System.currentTimeMillis();
		// Slave向Master汇报Offset，汇报到哪里
		private long currentReportedOffset = 0;
		// 标记从byteBufferRead变量读取数据的位置；初始化值为0
		private int dispatchPostion = 0;
		// 从Master接收数据Buffer
		private ByteBuffer byteBufferRead = ByteBuffer.allocate(ReadMaxBufferSize);
		private ByteBuffer byteBufferBackup = ByteBuffer.allocate(ReadMaxBufferSize);

		public HAClient() throws IOException {
			this.selector = RemotingUtil.openSelector();
		}

		public void updateMasterAddress(final String newAddr) {
			String currentAddr = this.masterAddress.get();
			if (currentAddr == null || !currentAddr.equals(newAddr)) {
				this.masterAddress.set(newAddr);
				log.info("update master address, OLD: " + currentAddr + " NEW: " + newAddr);
			}
		}
		/**
		 * 检查上次写入时间戳lastWriteTimestamp距离现在是否已经过了5秒，即每隔5秒向主用Broker进行一次物理偏移量报告（HAClient.currentReportedOffset）
		 * @return
		 */
		private boolean isTimeToReportOffset() {
			long interval = HAService.this.defaultMessageStore.getSystemClock().now() - this.lastWriteTimestamp;
			boolean needHeart = (interval > HAService.this.defaultMessageStore.getMessageStoreConfig().getHaSendHeartbeatInterval());

			return needHeart;
		}

		private boolean reportSlaveMaxOffset(final long maxOffset) {
			this.reportOffset.position(0);
			this.reportOffset.limit(8);
			this.reportOffset.putLong(maxOffset);
			this.reportOffset.position(0);
			this.reportOffset.limit(8);

			for (int i = 0; i < 3 && this.reportOffset.hasRemaining(); i++) {
				try {
					this.socketChannel.write(this.reportOffset);
				} catch (IOException e) {
					log.error(this.getServiceName() + "reportSlaveMaxOffset this.socketChannel.write exception", e);
					return false;
				}
			}

			return !this.reportOffset.hasRemaining();
		}

		// private void reallocateByteBuffer() {
		// ByteBuffer bb = ByteBuffer.allocate(ReadMaxBufferSize);
		// int remain = this.byteBufferRead.limit() - this.dispatchPostion;
		// bb.put(this.byteBufferRead.array(), this.dispatchPostion, remain);
		// this.dispatchPostion = 0;
		// this.byteBufferRead = bb;
		// }

		/**
		 * Buffer满了以后，重新整理一次
		 */
		private void reallocateByteBuffer() {
			// 检查byteBufferRead变量中的二进制数据是否解析完了（reamin=ReadMaxBufferSize-dispatchPostion）
			int remain = ReadMaxBufferSize - this.dispatchPostion;
			// 若reamin>0表示没有解析完，则将剩下的数据复制到HAClient.byteBufferBackup变量中；
			if (remain > 0) {
				this.byteBufferRead.position(this.dispatchPostion);

				this.byteBufferBackup.position(0);
				this.byteBufferBackup.limit(ReadMaxBufferSize);
				this.byteBufferBackup.put(this.byteBufferRead);
			}
			// 将byteBufferRead和byteBufferBackup的数据进行交换
			this.swapByteBuffer();

			// 重新初始化byteBufferRead变量的position等于reamin，即表示byteBufferRead中写入到了位置position
			this.byteBufferRead.position(remain);
			this.byteBufferRead.limit(ReadMaxBufferSize);
			this.dispatchPostion = 0;
		}

		private void swapByteBuffer() {
			ByteBuffer tmp = this.byteBufferRead;
			this.byteBufferRead = this.byteBufferBackup;
			this.byteBufferBackup = tmp;
		}

		private boolean processReadEvent() {
			int readSizeZeroTimes = 0;
			// 一直循环的读取并解析数据，直到HAClient.byteBufferRead:ByteBuffer中无可写的空间为止
			// 该byteBufferRead变量是在初始化时HAClient是创建的，初始化空间为ReadMaxBufferSize=4G
			while (this.byteBufferRead.hasRemaining()) {
				try {
					int readSize = this.socketChannel.read(this.byteBufferRead);
					if (readSize > 0) {
						// 若读取到数据，则首先更新HAClient.lastWriteTimestamp变量
						lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
						readSizeZeroTimes = 0;
						// 调用HAClient.dispatchReadRequest()方法对数据解析和处理，在dispatchReadRequest方法中循环的读取byteBufferRead变量中的数据
						boolean result = this.dispatchReadRequest();
						if (!result) {
							log.error("HAClient, dispatchReadRequest error");
							return false;
						}
					} else if (readSize == 0) {
						// 若为空将重复读取3次后仍然没有数据则跳出该循环
						if (++readSizeZeroTimes >= 3) {
							break;
						}
					} else {
						// TODO ERROR
						log.info("HAClient, processReadEvent read socket < 0");
						return false;
					}
				} catch (IOException e) {
					log.info("HAClient, processReadEvent read socket exception", e);
					return false;
				}
			}

			return true;
		}

		private boolean dispatchReadRequest() {
			final int MSG_HEADER_SIZE = 8 + 4; // phyoffset + size
			// 从SocketChannel中收到的数据的最后位置
			int readSocketPos = this.byteBufferRead.position();

			while (true) {
				int diff = this.byteBufferRead.position() - this.dispatchPostion;
				// 比较position减dispatchPostion的值大于12（消息头部长度为12个字节）
				// 若大于12个字节表示有心跳消息从主用Broker发送过来
				if (diff >= MSG_HEADER_SIZE) {
					// 在byteBufferRead中从dispatchPostion位置开始读取数据，初始化状态下dispatchPostion等于0；
					// 读取8个字节的数据即为主用Broker的同步的起始物理偏移量masterPhyOffset
					long masterPhyOffset = this.byteBufferRead.getLong(this.dispatchPostion);
					// 再后4字节为数据的大小bodySize
					int bodySize = this.byteBufferRead.getInt(this.dispatchPostion + 8);

					// 从备用Broker中获取最大的物理偏移量
					long slavePhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();

					// 发生重大错误
					if (slavePhyOffset != 0) {
						// 若与主用Broker传来的起始物理偏移量masterPhyOffset不相等，则直接返回false;
						if (slavePhyOffset != masterPhyOffset) {
							log.error("master pushed offset not equal the max phy offset in slave, SLAVE: " + slavePhyOffset + " MASTER: " + masterPhyOffset);
							return false;
						}
					}

					// 可以凑够一个请求
					// 若position-dispatchPostion的值大于消息头部长度12字节加上bodySize之和；则说明有数据同步，
					// 则继续在byteBufferRead中以position+dispatchPostion开始位置读取bodySize大小的数据
					if (diff >= (MSG_HEADER_SIZE + bodySize)) {
						byte[] bodyData = new byte[bodySize];
						this.byteBufferRead.position(this.dispatchPostion + MSG_HEADER_SIZE);
						this.byteBufferRead.get(bodyData);

						// TODO 结果是否需要处理，暂时不处理
						// 进行数据的写入
						HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);
						// 将byteBufferRead变量的position值重置为readSocketPos
						this.byteBufferRead.position(readSocketPos);
						// dispatchPostion值累计12+bodySize
						this.dispatchPostion += MSG_HEADER_SIZE + bodySize;

						// 检查当前备用Broker的最大物理偏移量是否大于了上次向主用Broker报告时的最大物理偏移量（HAClient.currentReportedOffset），
						// 若大于则更新HAClient.currentReportedOffset的值，并将最新的物理偏移量向主用Broker报告
						if (!reportSlaveMaxOffsetPlus()) {
							return false;
						}

						continue;
					}
				}

				// 若小于12个字节，并且byteBufferRead变量中没有可写空间（this.position>=this.limit）,则调用HAClient.reallocateByteBuffer()方法进行ByteBuffer的整理
				if (!this.byteBufferRead.hasRemaining()) {
					this.reallocateByteBuffer();
				}

				break;
			}

			return true;
		}

		private boolean reportSlaveMaxOffsetPlus() {
			boolean result = true;
			// 只要本地有更新，就汇报最大物理Offset
			long currentPhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
			// 检查当前备用Broker的最大物理偏移量是否大于了上次向主用Broker报告时的最大物理偏移量（HAClient.currentReportedOffset）
			if (currentPhyOffset > this.currentReportedOffset) {
				// 若大于则更新HAClient.currentReportedOffset的值，并将最新的物理偏移量向主用Broker报告。
				this.currentReportedOffset = currentPhyOffset;
				result = this.reportSlaveMaxOffset(this.currentReportedOffset);
				if (!result) {
					this.closeMaster();
					log.error("HAClient, reportSlaveMaxOffset error, " + this.currentReportedOffset);
				}
			}

			return result;
		}

		private boolean connectMaster() throws ClosedChannelException {
			if (null == socketChannel) {
				// 在备用Broker向Name Server注册时会返回主用Broker的地址
				String addr = this.masterAddress.get();
				if (addr != null) {

					SocketAddress socketAddress = RemotingUtil.string2SocketAddress(addr);
					if (socketAddress != null) {
						// 若有主用Broker地址则与主用Broker建立Socket链接
						this.socketChannel = RemotingUtil.connect(socketAddress);
						if (this.socketChannel != null) {
							// 在该链接上注册OP_READ操作，即监听主用Broker返回的消息
							this.socketChannel.register(this.selector, SelectionKey.OP_READ);
						}
					}
				}

				// 每次连接时，要重新拿到最大的Offset
				// 获取备用Broker本地的最大写入位置即最大物理偏移量，然后赋值给HAClient.currentReportedOffset变量
				this.currentReportedOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
				// 更新最后写入时间戳lastWriteTimestamp
				this.lastWriteTimestamp = System.currentTimeMillis();
			}

			return this.socketChannel != null;
		}

		private void closeMaster() {
			if (null != this.socketChannel) {
				try {

					SelectionKey sk = this.socketChannel.keyFor(this.selector);
					if (sk != null) {
						sk.cancel();
					}

					this.socketChannel.close();

					this.socketChannel = null;
				} catch (IOException e) {
					log.warn("closeMaster exception. ", e);
				}

				this.lastWriteTimestamp = 0;
				this.dispatchPostion = 0;

				this.byteBufferBackup.position(0);
				this.byteBufferBackup.limit(ReadMaxBufferSize);

				this.byteBufferRead.position(0);
				this.byteBufferRead.limit(ReadMaxBufferSize);
			}
		}

		@Override
		public void run() {
			log.info(this.getServiceName() + " service started");

			while (!this.isStoped()) {
				try {
					// 与主用Broker建立Socket链接，并在该链接上注册OP_READ操作，即监听主用Broker返回的消息
					if (this.connectMaster()) {
						// 先汇报最大物理Offset || 定时心跳方式汇报
						if (this.isTimeToReportOffset()) {
							// 若超过了5秒，则备用Broker向主用Broker报告备用Broker的当前最大物理偏移量的值，该消息只有8个字节，即为物理偏移量的值
							boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);
							if (!result) {
								this.closeMaster();
							}
						}

						// 等待应答
						// 该服务线程等待1秒钟，然后从SocketChannel读取主用Broker返回的数据
						this.selector.select(1000);

						// 接收数据
						// 然后从SocketChannel读取主用Broker返回的数据
						boolean ok = this.processReadEvent();
						if (!ok) {
							this.closeMaster();
						}

						// 只要本地有更新，就汇报最大物理Offset
						// 检查当前备用Broker的最大物理偏移量是否大于了上次向主用Broker报告时的最大物理偏移量（HAClient.currentReportedOffset），
						// 若大于则更新HAClient.currentReportedOffset的值，并将最新的物理偏移量向主用Broker报告。
						if (!reportSlaveMaxOffsetPlus()) {
							continue;
						}

						// 检查Master的反向心跳
						// 检查最后写入时间戳lastWriteTimestamp距离当前的时间
						long interval = HAService.this.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;
						// 若大于了5秒，表示这期间未收到过主用Broker的消息
						if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {
							log.warn("HAClient, housekeeping, found this connection[" + this.masterAddress + "] expired, " + interval);
							// 关闭与主用Broker的连接
							this.closeMaster();
							log.warn("HAClient, master not response some time, so close connection");
						}
					} else {
						this.waitForRunning(1000 * 5);
					}
				} catch (Exception e) {
					log.warn(this.getServiceName() + " service has exception. ", e);
					this.waitForRunning(1000 * 5);
				}
			}

			log.info(this.getServiceName() + " service end");
		}

		//
		// private void disableWriteFlag() {
		// if (this.socketChannel != null) {
		// SelectionKey sk = this.socketChannel.keyFor(this.selector);
		// if (sk != null) {
		// int ops = sk.interestOps();
		// ops &= ~SelectionKey.OP_WRITE;
		// sk.interestOps(ops);
		// }
		// }
		// }
		//
		//
		// private void enableWriteFlag() {
		// if (this.socketChannel != null) {
		// SelectionKey sk = this.socketChannel.keyFor(this.selector);
		// if (sk != null) {
		// int ops = sk.interestOps();
		// ops |= SelectionKey.OP_WRITE;
		// sk.interestOps(ops);
		// }
		// }
		// }

		@Override
		public String getServiceName() {
			return HAClient.class.getSimpleName();
		}
	}

	public AtomicLong getPush2SlaveMaxOffset() {
		return push2SlaveMaxOffset;
	}
}
